package com.warren.financial.lease.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.warren.financial.lease.realtime.common.FinancialLeaseCommon;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @AUTHOR warren
 * @date 2023/12/11
 **/
public class HBaseUtil {

    public static Connection getHBaseConnection(){
        Configuration conf = HBaseConfiguration.create();
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM,FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM_HOST);
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT,FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT_VALUE);

        try {
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void closeHBaseConnection(Connection conn){
        if (conn != null && !conn.isClosed()){
            try {
                conn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void createTable(Connection connection, String hbaseNamespace, String tableName, String... familyNames) throws IOException {

        if(familyNames == null || familyNames.length == 0){
            throw new RuntimeException("列族至少有一个");
        }
        Admin admin = connection.getAdmin();
        try {
            if(isNamespaceExist(hbaseNamespace,admin)){
                System.out.println("该命名空间未创建,创建命名空间"+hbaseNamespace);
            };

            TableName tableName1 = TableName.valueOf(hbaseNamespace, tableName);
            if(admin.tableExists(tableName1)){
                System.out.println("表已经存在" + hbaseNamespace + ":" + tableName);
            }else {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName1);

                for(String familyName : familyNames){
                    ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName));
                    tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
                }

                admin.createTable(tableDescriptorBuilder.build());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        admin.close();
    }

    public static void deleteTable(Connection hBaseConnection, String hbaseNamespace, String tableName) throws IOException {
        Admin admin = hBaseConnection.getAdmin();
        TableName tableName1 = TableName.valueOf(hbaseNamespace, tableName);

        try {
            admin.disableTable(tableName1);
            admin.deleteTable(tableName1);
        } catch (IOException e) {
            e.printStackTrace();
        }

        admin.close();
    }


    public static boolean isNamespaceExist(String namespace, Admin admin) throws IOException {
        NamespaceDescriptor namespaceDescriptor = null;
        try {
             namespaceDescriptor = admin.getNamespaceDescriptor(namespace);
        } catch (NamespaceNotFoundException e) {
            namespaceDescriptor = NamespaceDescriptor.create(namespace).build();

            admin.createNamespace(namespaceDescriptor);
        }
        return namespaceDescriptor == null;
    }

    public static void deleteRow(Connection hBaseConnection, String hbaseNamespace, String sinkTable, String rowKeyValue) throws IOException {

        Table table = hBaseConnection.getTable(TableName.valueOf(hbaseNamespace, sinkTable));

        try {
            table.delete(new Delete(Bytes.toBytes(rowKeyValue)));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void putRow(Connection hBaseConnection, String hbaseNamespace, String sinkTable, String rowKeyValue, String sinkFamily, String[] columns, String[] values) throws IOException {
        Table table = hBaseConnection.getTable(TableName.valueOf(hbaseNamespace, sinkTable));
        Put put = new Put(Bytes.toBytes(rowKeyValue));

        for (int i = 0; i < columns.length; i++) {
            if (values[i] == null){
                values[i] = "";
            }
            put.addColumn(Bytes.toBytes(sinkFamily),Bytes.toBytes(columns[i]),Bytes.toBytes(values[i]));
        }

        try {
            table.put(put);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static AsyncConnection getHbaseAsyncConnection() throws ExecutionException, InterruptedException {
        Configuration conf = HBaseConfiguration.create();
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM,FinancialLeaseCommon.HBASE_ZOOKEEPER_QUORUM_HOST);
        conf.set(FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT,FinancialLeaseCommon.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT_VALUE);

        return ConnectionFactory.createAsyncConnection(conf).get();
    }

    public static void closeHBaseAsyncConnection(AsyncConnection conn){
        if (conn != null && !conn.isClosed()){
            try {
                conn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 异步读取base中的维度数据
     * @param hBaseAsyncConnection
     * @param hbaseNamespace
     * @param tableName
     * @param rowKey
     * @return
     */
    public static JSONObject asyncReadDim(AsyncConnection hBaseAsyncConnection, String hbaseNamespace, String tableName, String rowKey){

        AsyncTable<AdvancedScanResultConsumer> table = hBaseAsyncConnection.getTable(TableName.valueOf(hbaseNamespace, tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        CompletableFuture<Result> result = table.get(get);
        JSONObject dim = new JSONObject();

        try {
            Result rs = result.get();

            for (Cell cell : rs.rawCells()){
                String column = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell));
                dim.put(column,value);
            }
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }

        return dim;
    }
}
